package com.test.study.ToolsUtils.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.Test;



/**
 * 本类用于kafka的生产者测试
 * 不同版本的Kafka使用不一样这里使用的此时版本是2.1.1
 * @author huyong 740 1005
 * @param <V>
 *
 */
public class KafkaProducerServer {

	@Test
	public void test() {
		String topic="hello";
		String message="my name is huyong";
		this.producerServer(topic, message);
	}
/**
 * 个人手动实现    send方法是异步的
 * huyong
 * @param topic
 * @param message
 * 2019年3月9日
 */
	public void producerServer(String topic, String message) {
		Properties properties = new Properties();
		KafkaProducer<String, String> kafkaProducer =null;
		try {
			// 1.首先准备生产的所有配置项
			properties.load(this.getClass().getClassLoader().getResourceAsStream("producer.properties"));
			// 2.创建一个生产者对象
			kafkaProducer= new KafkaProducer<String, String>(properties);
			// 3.装载消息：
			// 发送单个的
			ProducerRecord<String, String> record=new ProducerRecord<String,String>(topic,message);
			kafkaProducer.send(record);
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (kafkaProducer != null) {
				kafkaProducer.close();
			}
		}
	}
	
	/**
	 * 本类是官方文档  2.1.1给出的案例
	 * huyong
	 * 2019年3月9日
	 * 如果请求失败，则生产者可以自动重试，但由于我们已将重试指定为0，因此不会重试。启用重试也会打开重复的可能性
	 * delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
	 */
	@Test
	public void producerServer01() {
		 Properties props = new Properties();
		 props.put("bootstrap.servers", "localhost:9092");
		 props.put("acks", "all");//acks配置控制认为请求完成的标准。我们指定的“all”设置将导致完全提交记录时阻塞，这是最慢但最持久的设置。
		 props.put("delivery.timeout.ms", 300000);
		 props.put("batch.size", 16384);//生产者为每个分区维护未发送记录的缓冲区。这些缓冲区的大小由batch.size配置指定。使这个更大可以导致更多的批处理，但需要更多的内存
		 props.put("linger.ms", 1);//默认情况下，即使缓冲区中有其他未使用的空间，也可以立即发送缓冲区。但是，如果您想减少请求数量，可以将linger.ms设置为大于0的值。这将指示生产者在发送请求之前等待该毫秒数，希望更多记录到达以填满同一批次。这类似于TCP中的Nagle算法。
		 //在上面的代码片段中，由于我们将逗留时间设置为1毫秒，因此可能会在单个请求中发送所有100条记录。但是，如果我们没有填满缓冲区，此设置将为我们的请求添加1毫秒的延迟，等待更多记录到达。请注意，即使在linger.ms = 0的情况下，及时到达的记录通常也会一起批处理，因此在重负载下，无论延迟配置如何，都会发生批处理;但是，将此设置为大于0的值可以在不受最大负载影响的情况下以较少的延迟为代价导致更少，更有效的请求。
		 props.put("buffer.memory", 33554432);//buffer.memory控制生产者可用于缓冲的总内存量。如果记录的发送速度快于传输到服务器的速度，则此缓冲区空间将耗尽。当缓冲区空间耗尽时，其他发送调用将被阻止。阻塞时间的阈值由max.block.ms确定，之后它会抛出TimeoutException。
		 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		 props.put("partitioner.class","com.test.study.ToolsUtils.kafka.MyPartitioner");//自定义partition
		 //key.serializer和value.serializer指示如何将用户提供的键和值对象及其ProducerRecord转换为字节。您可以将包含的ByteArraySerializer或StringSerializer用于简单的字符串或字节类型。
		 Producer<String, String> producer = new KafkaProducer<String,String>(props);
		 for (int i = 0; i < 100; i++)
		     producer.send(new ProducerRecord<String, String>("hello", Integer.toString(i), Integer.toString(i)));

		 producer.close();
		 //从Kafka 0.11开始，KafkaProducer支持另外两种模式：幂等生成器和事务生成器。幂等生产者将卡夫卡的交付语义从至少一次加强到一次交付。特别是生产者重试将不再引入重复。事务生成器允许应用程序以原子方式将消息发送到多个分区（和主题！）。
		 //要启用幂等性，必须将enable.idempotence配置设置为true。如果设置，则重试配置将默认为Integer.MAX_VALUE，并且acks配置将默认为all。幂等生成器没有API更改，因此不需要修改现有应用程序以利用此功能。
		 
		
	}
	/**
	 * 使用幂等生成器
	 * 要利用幂等生成器，必须避免应用程序级别重新发送，因为这些不能重复数据删除。因此，如果应用程序启用幂等性，建议保留重试配置未设置，因为它将默认为Integer.MAX_VALUE。此外，如果send（ProducerRecord）即使无限次重试也会返回错误（例如，如果消息在发送之前在缓冲区中到期），则建议关闭生产者并检查上一次生成的消息的内容以确保它没有重复。最后，生产者只能保证在单个会话中发送的消息的幂等性。
	 *要使用事务生成器和话务员API，必须设置transactional.id配置属性。如果设置了transactional.id，则会自动启用幂等性以及幂等性所依赖的生产者配置。此外，交易中包含的主题应配置为持久性。特别是，replication.factor应该至少为3，并且这些主题的min.insync.replicas应该设置为2.最后，为了从端到端实现事务保证，消费者必须是配置为只读取已提交的消息。
     *transactional.id的目的是在单个生产者实例的多个会话中启用事务恢复。它通常从分区的有状态应用程序中的分片标识符派生。因此，对于在分区应用程序中运行的每个生产者实例，它应该是唯一的。
     *所有新的事务API都是阻塞的，并且会在失败时抛出异常。下面的示例说明了如何使用新API。它类似于上面的示例，除了所有100条消息都是单个事务的一部分。
	 * huyong
	 * 2019年3月9日
	 */
	@Test
	public void producerServer02() {
		 Properties props = new Properties();
		 props.put("bootstrap.servers", "localhost:9092");
		 props.put("transactional.id", "my-transactional-id");
		 Producer<String, String> producer = new KafkaProducer<String,String>(props, new StringSerializer(), new StringSerializer());

		 producer.initTransactions();

		 try {
		     producer.beginTransaction();
		     for (int i = 0; i < 100; i++)
		         producer.send(new ProducerRecord<String,String>("hello", Integer.toString(i), Integer.toString(i)));
		     producer.commitTransaction();
		 } catch (Exception e) {
		     // We can't recover from these exceptions, so our only option is to close the producer and exit.
		     producer.close();
		 } 
		 producer.close();
		
	}
	
	/**
	 * 正如在示例中暗示的那样，每个生产者只能有一个开放交易。在beginTransaction（）和commitTransaction（）调用之间发送的所有消息都将是单个事务的一部分。指定transactional.id时，生产者发送的所有消息都必须是事务的一部分。
	事务生成器使用异常来传递错误状态。特别是，不需要为返回的Future指定producer.send（）或调用.get（）的回调：如果任何producer.send（）或事务调用在期间遇到不可恢复的错误，将抛出KafkaException。交易。有关检测事务发送错误的更多详细信息，请参阅send（ProducerRecord）文档。
	通过在收到KafkaException时调用producer.abortTransaction（），我们可以确保将任何成功的写入标记为已中止，从而保持事务性保证。
	此客户端可以与版本0.10.0或更高版本的代理进行通信。较旧或较新的经纪人可能不支持某些客户端功能。例如，事务API需要代理版本0.11.0或更高版本。调用正在运行的代理版本中不可用的API时，您将收到UnsupportedVersionException。
	 */

}
